// Licensed to the Apache Software Foundation (ASF) under one or more
// contributor license agreements.  See the NOTICE file distributed with
// this work for additional information regarding copyright ownership.
// The ASF licenses this file to You under the Apache License, Version 2.0
// (the "License"); you may not use this file except in compliance with
// the License.  You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use anyhow::anyhow;
use once_cell::sync::Lazy;
use skywalking::reporter::{CollectItem, Report};
use std::{
    io::{self, Write, Read},
    ops::DerefMut,
    os::unix::net::UnixStream,
    path::{Path, PathBuf},
    sync::Mutex,
    time::{Duration, Instant},
};
use tracing::{debug, error, info, warn};

fn channel_send<T>(data: CollectItem, mut sender: T) -> anyhow::Result<()>
where
    T: DerefMut<Target = UnixStream>,
{
    let content = bincode::serde::encode_to_vec(&data, bincode::config::standard())?;

    sender.write_all(&content.len().to_le_bytes())?;
    sender.write_all(&content)?;
    sender.flush()?;

    Ok(())
}

struct ConnectionState {
    stream: Option<UnixStream>,
    reconnect_attempts: u32,
    last_attempt: Instant,
    // 是否已经成功连接过
    ever_connected: bool,
}

pub struct Reporter {
    worker_addr: PathBuf,
    state: Mutex<ConnectionState>,
    // 最大重连间隔（秒）
    max_backoff_secs: u64,
    // 连接超时（毫秒）
    connect_timeout_ms: u64,
    // 是否启用Keep-alive
    enable_keepalive: bool,
}

impl Reporter {
    pub fn new(worker_addr: impl AsRef<Path>) -> Self {
        Self {
            worker_addr: worker_addr.as_ref().to_path_buf(),
            state: Mutex::new(ConnectionState {
                stream: None,
                reconnect_attempts: 0,
                last_attempt: Instant::now(),
                ever_connected: false,
            }),
            max_backoff_secs: 30, // 最大退避时间30秒
            connect_timeout_ms: 1000, // 默认连接超时1秒
            enable_keepalive: true, // 默认启用keep-alive
        }
    }

    // 设置连接超时时间
    pub fn with_connect_timeout(mut self, timeout_ms: u64) -> Self {
        self.connect_timeout_ms = timeout_ms;
        self
    }

    // 设置是否启用Keep-alive
    pub fn with_keepalive(mut self, enable: bool) -> Self {
        self.enable_keepalive = enable;
        self
    }

    // 设置最大退避时间（秒）
    pub fn with_max_backoff(mut self, max_backoff_secs: u64) -> Self {
        self.max_backoff_secs = max_backoff_secs;
        self
    }

    // 计算指数退避时间
    fn calculate_backoff_time(&self, attempts: u32) -> Duration {
        if attempts == 0 {
            return Duration::from_millis(0);
        }
        
        // 指数退避: 2^attempts 毫秒，但最大不超过 max_backoff_secs 秒
        let base_ms = 1u64 << attempts.min(16); // 防止溢出
        let backoff_ms = base_ms.min(self.max_backoff_secs * 1000);
        Duration::from_millis(backoff_ms)
    }

    // 检查连接是否可用
    fn is_connection_alive(&self, stream_opt: &Option<UnixStream>) -> bool {
        if let Some(stream) = stream_opt {
            debug!("Checking if connection is alive");
            
            // 方法1：先检查peer_addr，这是基本检查
            match stream.peer_addr() {
                Ok(_) => {
                    debug!("Connection peer_addr check passed, performing additional checks");
                    
                    // 方法2：尝试进行简单的读写操作来验证连接真正可用
                    // 对于Unix域套接字，可以设置非阻塞模式，尝试读取而不是peek（因为peek是不稳定特性）
                    let mut stream_clone = match stream.try_clone() {
                        Ok(clone) => clone,
                        Err(e) => {
                            warn!(?e, "Failed to clone stream for connection check");
                            return false;
                        }
                    };
                    
                    if let Err(e) = stream_clone.set_nonblocking(true) {
                        warn!(?e, "Failed to set nonblocking mode for connection check");
                        return false;
                    }
                    
                    // 创建一个小的缓冲区来检查连接
                    let mut probe_result = true;
                    
                    // 尝试读取数据（非阻塞）
                    // 如果返回Ok(0)，表示连接开放但连接已关闭（EOF）
                    // 如果返回WouldBlock，表示连接开放但暂无数据可读（正常情况）
                    // 其他错误可能表示连接有问题
                    let mut buf = [0u8; 1];
                    match stream_clone.read(&mut buf) {
                        Ok(0) => {
                            // 读取到EOF，表示连接已关闭
                            debug!("Connection read returned EOF, connection is closed");
                            probe_result = false;
                        },
                        Ok(_) => {
                            // 读取到数据，连接正常但可能有未处理的数据
                            debug!("Connection read operation returned data, connection is alive");
                        },
                        Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                            // 这是预期的错误，表示连接正常但没有数据可读
                            debug!("Connection read returned WouldBlock, which is expected for a healthy connection");
                        },
                        Err(e) => {
                            // 其他错误可能表示连接有问题
                            debug!(?e, "Connection read operation failed with unexpected error");
                            probe_result = false;
                        }
                    }
                    
                    // 恢复阻塞模式
                    if let Err(e) = stream_clone.set_nonblocking(false) {
                        warn!(?e, "Failed to reset blocking mode after connection check");
                    }
                    
                    if probe_result {
                        debug!("All connection checks passed, connection is alive");
                        return true;
                    } else {
                        debug!("Additional connection checks failed");
                        // 尝试关闭已损坏的连接
                        if let Err(close_err) = stream_clone.shutdown(std::net::Shutdown::Both) {
                            debug!(error = ?close_err, "Error shutting down detected broken connection");
                        } else {
                            debug!("Successfully shut down detected broken connection");
                        }
                        return false;
                    }
                },
                Err(err) => {
                    match err.kind() {
                        io::ErrorKind::BrokenPipe | io::ErrorKind::ConnectionReset => {
                            debug!(error = ?err, "Connection is broken (BrokenPipe/ConnectionReset)");
                            // 尝试关闭已损坏的连接
                            if let Err(close_err) = stream.shutdown(std::net::Shutdown::Both) {
                                debug!(error = ?close_err, "Error shutting down broken connection");
                            } else {
                                debug!("Successfully shut down broken connection");
                            }
                            return false;
                        },
                        _ => {
                            warn!(?err, "Unexpected error checking connection status");
                            // 尝试关闭有问题的连接
                            if let Err(close_err) = stream.shutdown(std::net::Shutdown::Both) {
                                debug!(error = ?close_err, "Error shutting down problematic connection");
                            } else {
                                debug!("Successfully shut down problematic connection");
                            }
                            return false;
                        }
                    }
                }
            }
        }
        debug!("No connection to check");
        false
    }

    // 重新连接方法
    fn reconnect(&self) -> anyhow::Result<()> {
        // 获取锁并尝试重新连接
        debug!("Attempting to reconnect to worker");
        let mut state = self.state.lock().map_err(|_| anyhow!("Get state lock failed"))?;
        
        // 再次检查是否需要重连 (双重检查以避免多线程竞争)
        debug!("Double-checking if reconnection is needed");
        if let Some(ref stream) = state.stream {
            if let Ok(clone) = stream.try_clone() {
                if self.is_connection_alive(&Some(clone)) {
                    debug!("Connection is still alive, no need to reconnect");
                    return Ok(());
                }
                debug!("Connection check failed, need to reconnect");
            } else {
                debug!("Failed to clone stream for checking");
            }
            // 连接已失效，清除它
            debug!("Connection is invalid, clearing it");
            if let Some(old_stream) = state.stream.take() {
                let _ = old_stream.shutdown(std::net::Shutdown::Both);
                debug!("Cleared invalid connection");
            }
        } else {
            debug!("No existing connection in state");
        }
        
        // 检查是否需要等待退避时间
        debug!("Checking if in backoff period");
        let now = Instant::now();
        let time_since_last_attempt = now.duration_since(state.last_attempt);
        let backoff_time = self.calculate_backoff_time(state.reconnect_attempts);
        
        if time_since_last_attempt < backoff_time {
            let wait_time = backoff_time - time_since_last_attempt;
            debug!(
                attempts = state.reconnect_attempts,
                wait_time_ms = wait_time.as_millis(),
                "In backoff period"
            );
            return Err(anyhow!("In backoff period, {} ms remaining", wait_time.as_millis()));
        }
        
        debug!(
            ?self.worker_addr, 
            "Attempting to reconnect to Unix socket"
        );
        state.last_attempt = now;
        
        // 使用超时来创建连接
        debug!("Creating connection with timeout");
        match self.connect_with_timeout() {
            Ok(stream) => {
                debug!("Connection established, configuring socket options");
                // 设置一些有助于保持连接的选项
                if let Err(e) = stream.set_nonblocking(false) {
                    warn!(?e, "Failed to set blocking mode on socket");
                } else {
                    debug!("Set socket to blocking mode");
                }
                
                // 如果启用了Keep-alive，设置相关选项
                if self.enable_keepalive {
                    debug!("Configuring keep-alive options");
                    #[cfg(target_os = "linux")]
                    {
                        use std::os::unix::io::AsRawFd;
                        let fd = stream.as_raw_fd();
                        unsafe {
                            let val: libc::c_int = 1;
                            if libc::setsockopt(
                                fd,
                                libc::SOL_SOCKET,
                                libc::SO_KEEPALIVE,
                                &val as *const _ as *const libc::c_void,
                                std::mem::size_of_val(&val) as libc::socklen_t,
                            ) < 0 {
                                warn!("Failed to set SO_KEEPALIVE: {}", io::Error::last_os_error());
                            } else {
                                debug!("Successfully set SO_KEEPALIVE");
                            }
                            
                            // 设置TCP存活时间相关参数，减少无响应时间
                            #[cfg(target_os = "linux")]
                            {
                                debug!("Configuring TCP keep-alive parameters");
                                // TCP_KEEPIDLE: 空闲多久后开始发送keepalive探测包（秒）
                                let val: libc::c_int = 5;
                                if libc::setsockopt(
                                    fd,
                                    libc::IPPROTO_TCP,
                                    libc::TCP_KEEPIDLE,
                                    &val as *const _ as *const libc::c_void,
                                    std::mem::size_of_val(&val) as libc::socklen_t,
                                ) < 0 {
                                    debug!("Failed to set TCP_KEEPIDLE: {}", io::Error::last_os_error());
                                } else {
                                    debug!("Set TCP_KEEPIDLE to {} seconds", val);
                                }
                                
                                // TCP_KEEPINTVL: 探测包发送间隔（秒）
                                let val: libc::c_int = 1;
                                if libc::setsockopt(
                                    fd,
                                    libc::IPPROTO_TCP,
                                    libc::TCP_KEEPINTVL,
                                    &val as *const _ as *const libc::c_void,
                                    std::mem::size_of_val(&val) as libc::socklen_t,
                                ) < 0 {
                                    debug!("Failed to set TCP_KEEPINTVL: {}", io::Error::last_os_error());
                                } else {
                                    debug!("Set TCP_KEEPINTVL to {} seconds", val);
                                }
                                
                                // TCP_KEEPCNT: 没有响应时重试次数
                                let val: libc::c_int = 3;
                                if libc::setsockopt(
                                    fd,
                                    libc::IPPROTO_TCP,
                                    libc::TCP_KEEPCNT,
                                    &val as *const _ as *const libc::c_void,
                                    std::mem::size_of_val(&val) as libc::socklen_t,
                                ) < 0 {
                                    debug!("Failed to set TCP_KEEPCNT: {}", io::Error::last_os_error());
                                } else {
                                    debug!("Set TCP_KEEPCNT to {}", val);
                                }
                            }
                        }
                    }
                } else {
                    debug!("Keep-alive is disabled, skipping keep-alive configuration");
                }
                
                // 更新连接状态
                debug!("Updating connection state");
                state.reconnect_attempts = 0;
                if !state.ever_connected {
                    state.ever_connected = true;
                    info!(
                        ?self.worker_addr, 
                        "Successfully connected to skywalking-php-worker"
                    );
                } else {
                    info!(
                        ?self.worker_addr, 
                        "Successfully reconnected to skywalking-php-worker"
                    );
                }
                state.stream = Some(stream);
                debug!("Reconnection completed successfully");
                Ok(())
            }
            Err(err) => {
                // 增加重连尝试次数
                state.reconnect_attempts += 1;
                
                // 减少日志输出频率
                if state.reconnect_attempts == 1 || state.reconnect_attempts % 10 == 0 {
                    warn!(
                        ?err, 
                        attempts = state.reconnect_attempts,
                        next_attempt_in_ms = self.calculate_backoff_time(state.reconnect_attempts).as_millis(),
                        "Failed to connect to Unix socket, will retry with backoff"
                    );
                } else {
                    debug!(
                        ?err,
                        attempts = state.reconnect_attempts,
                        next_attempt_in_ms = self.calculate_backoff_time(state.reconnect_attempts).as_millis(),
                        "Connection attempt failed"
                    );
                }
                Err(err.into())
            }
        }
    }

    // 带超时的连接方法
    fn connect_with_timeout(&self) -> io::Result<UnixStream> {
        // Unix域套接字不支持标准的超时连接方式，需要使用非阻塞模式和轮询
        debug!("Connecting to Unix socket at {:?}", self.worker_addr);
        let socket = match UnixStream::connect(&self.worker_addr) {
            Ok(s) => {
                debug!("Initial connection to Unix socket successful");
                s
            },
            Err(e) => {
                debug!(error = ?e, "Failed to connect to Unix socket");
                return Err(e);
            }
        };
        
        if self.connect_timeout_ms == 0 {
            debug!("Timeout is disabled, returning socket immediately");
            return Ok(socket); // 不设置超时，直接返回
        }
        
        // 设置非阻塞模式
        debug!("Setting socket to non-blocking mode for timeout handling");
        if let Err(e) = socket.set_nonblocking(true) {
            debug!(error = ?e, "Failed to set non-blocking mode");
            return Err(e);
        }
        
        // 使用select/poll等待连接完成或超时
        debug!("Waiting for connection to complete with timeout of {}ms", self.connect_timeout_ms);
        let start = Instant::now();
        let timeout = Duration::from_millis(self.connect_timeout_ms);
        
        loop {
            match socket.peer_addr() {
                Ok(_) => {
                    // 连接成功，恢复阻塞模式
                    debug!("Connection confirmed successful, resetting to blocking mode");
                    if let Err(e) = socket.set_nonblocking(false) {
                        debug!(error = ?e, "Failed to reset to blocking mode");
                        return Err(e);
                    }
                    debug!("Connection established successfully");
                    return Ok(socket);
                }
                Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                    // 连接仍在进行中，检查是否超时
                    let elapsed = start.elapsed();
                    if elapsed > timeout {
                        debug!(
                            elapsed_ms = elapsed.as_millis(),
                            timeout_ms = timeout.as_millis(),
                            "Connection timed out"
                        );
                        return Err(io::Error::new(
                            io::ErrorKind::TimedOut,
                            "Connection timed out",
                        ));
                    }
                    
                    // 短暂休眠避免占用CPU
                    if elapsed.as_millis() % 100 == 0 {
                        debug!(
                            elapsed_ms = elapsed.as_millis(),
                            timeout_ms = timeout.as_millis(),
                            "Still waiting for connection"
                        );
                    }
                    std::thread::sleep(Duration::from_millis(10));
                }
                Err(e) => {
                    debug!(error = ?e, "Error checking connection status");
                    return Err(e); // 其他错误
                }
            }
        }
    }

    fn get_or_create_stream(&self) -> anyhow::Result<UnixStream> {
        // 先检查是否有有效的连接
        debug!("Checking if existing connection is valid");
        {
            let state = self.state.lock().map_err(|_| anyhow!("Get state lock failed"))?;
            if let Some(ref stream) = state.stream {
                if self.is_connection_alive(&Some(stream.try_clone()?)) {
                    debug!("Using existing valid connection");
                    return Ok(stream.try_clone()?);
                }
                debug!("Existing connection is invalid");
            } else {
                debug!("No existing connection found");
            }
        }
        
        // 需要重新连接
        debug!("Need to reconnect to worker");
        self.reconnect()?;
        
        // 获取连接
        debug!("Getting connection after reconnection");
        let state = self.state.lock().map_err(|_| anyhow!("Get state lock failed"))?;
        if let Some(ref stream) = state.stream {
            debug!("Successfully obtained connection after reconnection");
            Ok(stream.try_clone()?)
        } else {
            debug!("Failed to obtain connection even after reconnection attempt");
            Err(anyhow!("Failed to get stream after reconnection"))
        }
    }

    fn try_report(&self, item: CollectItem) -> anyhow::Result<()> {
        // 记录此次尝试的开始时间，用于检测重试是否过于频繁
        static LAST_ERROR_TIME: Lazy<Mutex<Instant>> = Lazy::new(|| Mutex::new(Instant::now()));
        static ERROR_COUNT: Lazy<Mutex<u32>> = Lazy::new(|| Mutex::new(0));
        
        debug!("Attempting to get connection for sending data");
        match self.get_or_create_stream() {
            Ok(mut stream) => {
                debug!("Got valid connection, sending data");
                match channel_send(item, &mut stream) {
                    Ok(_) => {
                        // 成功发送，重置错误计数
                        if let Ok(mut count) = ERROR_COUNT.lock() {
                            *count = 0;
                        }
                        debug!("Successfully sent data");
                        Ok(())
                    }
                    Err(e) => {
                        debug!("Send operation failed");
                        // 处理发送错误，尝试重置连接
                        if let Some(io_err) = e.downcast_ref::<io::Error>() {
                            debug!(
                                ?io_err,
                                error_kind = ?io_err.kind(),
                                "Error details during channel send"
                            );
                            
                            // 如果是连接错误，显式尝试重新连接一次
                            if io_err.kind() == io::ErrorKind::BrokenPipe || 
                               io_err.kind() == io::ErrorKind::ConnectionReset ||
                               io_err.kind() == io::ErrorKind::ConnectionRefused {
                                debug!("Connection error detected, will force reconnection");
                                // 只有当错误频率不高时才记录警告日志
                                let mut record_warning = true;
                                if let (Ok(mut last_time), Ok(mut count)) = (LAST_ERROR_TIME.lock(), ERROR_COUNT.lock()) {
                                    *count += 1;
                                    let now = Instant::now();
                                    // 如果一秒内错误超过5次，减少日志输出
                                    if now.duration_since(*last_time) < Duration::from_secs(1) && *count > 5 {
                                        if *count % 50 == 0 { // 每50次错误才记录一次日志
                                            debug!(
                                                error_count = *count,
                                                "High frequency of connection errors, reducing log output"
                                            );
                                        }
                                        record_warning = false;
                                    } else if now.duration_since(*last_time) >= Duration::from_secs(1) {
                                        // 超过1秒，重置计数
                                        *last_time = now;
                                        *count = 1;
                                    }
                                }
                                
                                if record_warning {
                                    warn!(
                                        ?io_err,
                                        error_kind = ?io_err.kind(),
                                        "Channel send failed, forcing reconnection"
                                    );
                                }
                                
                                // 强制关闭当前连接以触发重连
                                if let Ok(mut state) = self.state.lock() {
                                    if let Some(old_stream) = state.stream.take() {
                                        let _ = old_stream.shutdown(std::net::Shutdown::Both);
                                        debug!("Explicitly closed connection due to send error");
                                    } else {
                                        debug!("No connection to close in state");
                                    }
                                } else {
                                    debug!("Failed to acquire lock to close connection");
                                }
                            }
                        }
                        // 短暂延迟，避免在连接失败时立即重试造成CPU负载过高
                        debug!("Sleeping briefly before retrying");
                        std::thread::sleep(Duration::from_millis(10));
                        Err(e)
                    }
                }
            },
            Err(err) => {
                debug!("Failed to get valid connection: {}", err);
                // 只有当错误不是因为退避期而导致时，才需要延迟
                if !err.to_string().contains("In backoff period") {
                    // 短暂延迟，避免在连接失败时立即重试造成CPU负载过高
                    debug!("Sleeping briefly before retrying (not in backoff period)");
                    std::thread::sleep(Duration::from_millis(10));
                } else {
                    debug!("In backoff period, not sleeping additional time");
                }
                Err(err)
            }
        }
    }
}

impl Report for Reporter {
    fn report(&self, item: CollectItem) {
        if let Err(err) = self.try_report(item) {
            // 根据错误类型决定日志级别
            if err.to_string().contains("In backoff period") {
                // 在退避期内的错误，只有偶尔记录日志
                if let Ok(state) = self.state.lock() {
                    if state.reconnect_attempts % 100 == 1 {
                        debug!(
                            ?err, 
                            reconnect_attempts = state.reconnect_attempts,
                            backoff_ms = self.calculate_backoff_time(state.reconnect_attempts).as_millis(),
                            "Channel send skipped, in reconnection backoff period"
                        );
                    }
                }
            } else {
                match err.downcast_ref::<io::Error>() {
                    Some(io_err) if io_err.kind() == io::ErrorKind::BrokenPipe => {
                        // 对于断开的连接，使用warn级别
                        warn!(
                            ?err, 
                            error_kind = ?io_err.kind(),
                            "Channel send failed due to broken pipe, will attempt to reconnect"
                        );
                    }
                    Some(io_err) if io_err.kind() == io::ErrorKind::ConnectionReset => {
                        warn!(
                            ?err, 
                            error_kind = ?io_err.kind(),
                            "Channel send failed due to connection reset, will attempt to reconnect"
                        );
                    }
                    Some(io_err) if io_err.kind() == io::ErrorKind::ConnectionRefused => {
                        warn!(
                            ?err, 
                            error_kind = ?io_err.kind(),
                            "Connection refused, worker may be down, will attempt to reconnect"
                        );
                    }
                    _ => {
                        // 其他错误使用error级别
                        error!(?err, "Channel send failed");
                    }
                }
            }
        }
    }
}
